package com.yeziji.config.rabbitMq;

import cn.hutool.crypto.digest.MD5;
import com.yeziji.common.business.dataMessage.entity.DataMessageRecordEntity;
import com.yeziji.common.business.dataMessage.service.DataMessageRecordService;
import com.yeziji.common.rabbitMq.RabbitMqTopicHandler;
import com.yeziji.config.EnvHolder;
import com.yeziji.constant.DataSyncStatusEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Lazy;

import javax.annotation.Nonnull;
import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * rabbit mq 消息异常回调
 *
 * @author hwy
 * @since 2024/10/25 15:42
 **/
@Slf4j
public class RabbitMqReturnCallback implements RabbitTemplate.ReturnsCallback {
    @Lazy
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private DataMessageRecordService dataMessageRecordService;
    public static final Map<String, Integer> RETRY_MAP = new ConcurrentHashMap<>();
    private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(2);

    @Override
    public void returnedMessage(@Nonnull ReturnedMessage returnedMessage) {
        log.info("returnedMessage: {}", returnedMessage);
        int replyCode = returnedMessage.getReplyCode();
        String exchange = returnedMessage.getExchange();
        String routeKey = returnedMessage.getRoutingKey();
        // 判断是否是发送到指子客户端，如果是的话，这里如果 ack 为 true 说明消息已经传递到了(route 和 exchange 必须规范一致）
        int routerIndex = routeKey.indexOf('.');
        if (!routeKey.substring(routerIndex).equals(exchange.substring(routerIndex))) {
            String id = returnedMessage.getMessage().getMessageProperties().getCorrelationId();
            DataMessageRecordEntity record = dataMessageRecordService.getByFlagId(id);
            if (record != null) {
                record.setStatus(DataSyncStatusEnum.FAILED.getCode());
                dataMessageRecordService.updateById(record);
                return;
            }
        }

        // 通过处理器处理消息
        RabbitMqTopicHandler rabbitMqHandler = EnvHolder.getAsRabbitMqHandler(exchange, routeKey);
        byte[] body = returnedMessage.getMessage().getBody();
        log.warn("[{}]{}:{} ---> {}", replyCode, exchange, routeKey, new String(body));
        // 判断是否为重试
        if (replyCode == 312) {
            String flag = MD5.create().digestHex(body);
            String retryKey = exchange + ":" + routeKey + ":" + flag;
            Integer retryCount = RETRY_MAP.getOrDefault(retryKey, 0);
            if (retryCount < 5) {
                log.info("{} retry ...", retryCount);
                rabbitTemplate.convertAndSend(exchange, routeKey, body);
            } else {
                log.error("send message failed");
                // 当 retryCount 达到 5 时，开启线程延时 10 秒后从 RETRY_MAP 中移除
                SCHEDULED_EXECUTOR_SERVICE.schedule(() -> RETRY_MAP.remove(retryKey), 10, TimeUnit.SECONDS);
            }
            RETRY_MAP.put(retryKey, ++retryCount);
        } else if (rabbitMqHandler != null) {
            rabbitMqHandler.returned(returnedMessage);
        }
    }
}
